package com.hmeter.hadoop.mapreduce.jobtracker;

import java.io.IOException;
import java.text.DecimalFormat;
import java.util.Collection;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapred.ClusterStatus;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.JobStatus;
import org.apache.hadoop.mapreduce.Cluster.JobTrackerStatus;


public class JobTrackerMetrics {

	public JobTrackerMetrics() { };
	
	protected static JobClient prepareClient(String jobTracker) throws IOException
	{
		Configuration conf = new Configuration();
		conf.set("mapred.job.tracker", jobTracker);
		JobConf jobConf = new JobConf(conf);
		JobClient client = new JobClient(jobConf);
		return client;
	}
	
	public String getJobTrackerMetric(String jobTracker) throws IOException
	{
		StringBuilder jobTrackerResult = new StringBuilder();
		JobClient client = prepareClient(jobTracker);
		
		ClusterStatus clusterStatus = client.getClusterStatus();
		JobTrackerStatus jobTrackerStatus = clusterStatus.getJobTrackerStatus();
		
		//Task tracker info
		StringBuilder taskTrackersStatus = new StringBuilder();
		Collection<String> activeTaskTrackers = clusterStatus.getActiveTrackerNames();
		Collection<String> blackListedTaskTrackers = clusterStatus.getBlacklistedTrackerNames();
				
		taskTrackersStatus.append(" <tasktrackers>\n");
		
		for (String taskTracker : activeTaskTrackers)
		{
			taskTrackersStatus.append("  <tasktracker name='" + taskTracker + "' status='active'/>\n");
		}
		
		for (String taskTracker : blackListedTaskTrackers)
		{
			taskTrackersStatus.append("  <tasktracker name='" + taskTracker + "' status='blacklisted'/>\n");
		}
				
		taskTrackersStatus.append(" </tasktrackers>\n");
		
		//Cluster info
		StringBuilder clusterInfo = new StringBuilder();
		
		DecimalFormat twoDForm = new DecimalFormat("#.##");
		double activeMapTasks = Double.valueOf(clusterStatus.getMapTasks());
		double maxMapTasks = Double.valueOf(clusterStatus.getMaxMapTasks());
		double mapTasksUtilization = Double.valueOf(twoDForm.format(activeMapTasks/maxMapTasks)) * 100;
		
		double activeReduceTasks = Double.valueOf(clusterStatus.getReduceTasks());
		double maxReduceTasks = Double.valueOf(clusterStatus.getMaxReduceTasks());
		double reduceTasksUtilization = Double.valueOf(twoDForm.format(activeReduceTasks/maxReduceTasks)) * 100;
		
		clusterInfo.append(" <clusterinfo>\n");
		clusterInfo.append("  <maptasks>\n");
		clusterInfo.append("   <capacity>"+clusterStatus.getMaxMapTasks()+"</capacity>\n");
		clusterInfo.append("   <active>"+clusterStatus.getMapTasks()+"</active>\n");
		clusterInfo.append("   <utilization>"+mapTasksUtilization+"%</utilization>\n");
		clusterInfo.append("  </maptasks>\n");
		clusterInfo.append("  <reducetasks>\n");
		clusterInfo.append("   <capacity>"+clusterStatus.getMaxReduceTasks()+"</capacity>\n");
		clusterInfo.append("   <active>"+clusterStatus.getReduceTasks()+"</active>\n");
		clusterInfo.append("   <utilization>"+reduceTasksUtilization+"%</utilization>\n");
		clusterInfo.append("  </reducetasks>\n");
		clusterInfo.append("  <runningjobs>\n");
		
		JobStatus[] runningJobs = client.jobsToComplete();
		for (JobStatus runningJob : runningJobs)
		{
			clusterInfo.append("   <runningjob jobid='" + runningJob.getJobID().toString() + "'/>\n");
		}
		
		clusterInfo.append("  </runningjobs>\n");
		clusterInfo.append(" </clusterinfo>\n");
		
		jobTrackerResult.append("<jobtrackermetric jobtrackerstatus='" + jobTrackerStatus.toString() + "'>\n");
		jobTrackerResult.append(taskTrackersStatus);
		jobTrackerResult.append(clusterInfo);
		jobTrackerResult.append("</jobtrackermetric>");
				
		return jobTrackerResult.toString();
	}
}
